﻿namespace DataIntegration.Extracters
{
    using System;
    using System.Collections.Generic;
    using System.Data;
    using System.Reflection;
    using Entities;
    using Entities.Data;
    using Entities.Triggers;
    using log4net;
    using Messaging.Consumer;
    using Messaging.Publisher;
    using Storage;
    using Utils;
    using Utils.Serialization;

    public class CoalescingExtracter : Extracter
    {
        private static readonly ILog Logger = LogManager.GetLogger(MethodBase.GetCurrentMethod().ReflectedType);

        private readonly IConsumerFactory consumerFactory;

        private readonly List<string> dataTopics;
        
        private readonly int dataItemCount;

        private readonly List<IConsumer> consumers;

        private readonly IDictionary<string, List<ExtractedData>> stagingHeap;

        public CoalescingExtracter(
                string name, 
                IPublisherFactory publisherFactory, 
                IDataStoreFactory dataStoreFactory, 
                string filter,
                IConsumerFactory consumerFactory,
                List<string> dataTopics) 
            : base(name, publisherFactory, dataStoreFactory, filter)
        {
            this.consumerFactory = consumerFactory;
            this.dataTopics = dataTopics;
            this.dataItemCount = dataTopics.Count;
            this.stagingHeap = new Dictionary<string, List<ExtractedData>>();
            this.consumers = new List<IConsumer>();
        }

        public override void Dispose()
        {
            base.Dispose();
            Logger.DebugFormat("Disposing [{0}]", this.name);

            foreach (IConsumer consumer in this.consumers)
            {
                consumer.Dispose();
            }
        }

        protected override void Setup()
        {
            base.Setup();
            Logger.DebugFormat("Setting up [{0}]", this.name);
            IConsumer consumer = this.consumerFactory.GetConsumer(this.DaemonName, MessageType.Data, this.name);
            consumer.MessageReceived += this.MessageReceived;
            consumer.Start();

            foreach (string dataTopic in this.dataTopics)
            {
                string topic = string.Format("{0}.{1}.{2}", App.Name, MessageType.Data, dataTopic);
                consumer.ConsumeDurable(topic);
                this.consumers.Add(consumer);
            }
        }

        protected override void Extract(TriggerInstruction triggerInstruction)
        {
            List<ExtractedData> stagedData = this.stagingHeap[triggerInstruction.Target];
            
            foreach (ExtractedData extractedData in stagedData)
            {
                this.Commit(extractedData.Data);
            }
        }

        private void MessageReceived(string messageId, string message)
        {
            ExtractedData extractedData = SerializationFactory.DeserializeFromXml<ExtractedData>(message);

            if (!this.stagingHeap.ContainsKey(messageId))
            {
                lock (this.stagingHeap)
                {
                    if (!this.stagingHeap.ContainsKey(messageId))
                    {
                        this.stagingHeap.Add(messageId, new List<ExtractedData>());
                    }
                }
            }

            this.stagingHeap[messageId].Add(extractedData);
            
            if (this.stagingHeap[messageId].Count == this.dataItemCount)
            {
                this.ProcessTrigger(Guid.NewGuid().ToString(), new TriggerInstruction(messageId));
            }
        }
    }
}
